package org.demo.test.thread;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 设计一个阻塞队列，支持同步put生产对象，以及get消费对象；
 * 如果在大于队列容量时同步阻塞线程等待队列空间被消费后释放
 *
 */
public class BlockedQueue<T> {

    private List<T> list = new ArrayList();

    private Lock lock = new ReentrantLock();

    private Condition conditionEmpty;

    private Condition conditionNotEmpty;

    private int maxSize = 20;

    public BlockedQueue(int maxSize){
        conditionEmpty = lock.newCondition();
        conditionNotEmpty = lock.newCondition();
        this.maxSize = maxSize;
    }

    public void put(T o){
        lock.lock();
        try {
            // 判断当前队列是否达到上限，考虑到多线程同时在put的情况下，
            // 需要采用while在每次被唤醒是再进行一次判断
            while(list.size() >= maxSize){
                conditionEmpty.await();
            }
            list.add(o);
            conditionNotEmpty.signalAll();
        }catch (Exception e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    public T get(){
        lock.lock();
        T t = null;
        try {
            // 采用while 在多个现场等待获取时,如果被notify后，第一个线程拿走后，后续被唤醒的线程是拿不到需要再次等待的
            while (list.isEmpty()) {
                conditionNotEmpty.await();
            }
            t = list.remove(list.size() - 1);
            conditionEmpty.signalAll();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        return t;
    }

    public T get(long mills){
        lock.lock();
        T t = null;
        try {
            if(mills < 0) {
                // 采用while 在多个现场等待获取时,如果被notify后，第一个线程拿走后，后续被唤醒的线程是拿不到需要再次等待的
                while (list.isEmpty()) {
                    conditionNotEmpty.await();
                }
                t = list.remove(list.size() - 1);
                conditionEmpty.signalAll();
            }else {
                // 获取超时时刻
                long future = System.currentTimeMillis() + mills;
                long remaining = mills;
                while (list.isEmpty() && remaining > 0) {
                    // 等待指定超时时间
                    conditionNotEmpty.await(remaining, TimeUnit.MILLISECONDS);
                    // 如果被唤醒，则计算出剩余超时时间；主要由于被唤醒后不一定存在数据
                    remaining = future - System.currentTimeMillis();
                }

                // 判断list不为空则范围对应的对象，如果为空则任务是超时无法获取对象
                if(!list.isEmpty()) {
                    t = list.remove(list.size() - 1);
                    conditionEmpty.signalAll();
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        return t;
    }

    public static void main(String[] args) {
        BlockedQueue<String> queue = new BlockedQueue<>(5);
        int mode = 2; // 1:消费慢于生产，2生产慢于消费

        Thread t1 = new Thread(()->{
            while (true) {
//                Scanner scanner = new Scanner(System.in);
//                String line = scanner.nextLine();
                if(mode == 1) {
                    String line = "OK";
                    queue.put("IN:" + line);
                    System.out.println("put:" + line);
                }else if(mode == 2){
                    try {
                        Thread.sleep(10000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    String line = "OK";
                    queue.put("IN:" + line);
                    System.out.println("put:" + line);
                }
            }
        });

        t1.start();

        Thread t2 = new Thread(()->{
            while (true) {
                /**
                 * 消费满
                 */
                if(mode == 1) {
                    try {
                        Thread.sleep(10000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("输出：" + queue.get());
                }else if(mode == 2){
                    System.out.println("输出：" + queue.get());
                }
            }
        });

        t2.start();

        try {
            t1.join();
            t2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}
